/* 
  Copyright (c) SilverLining 2015
  Date:    2016-1-7
  File:    process_manager.cpp  
  Description: 

*/

#include "stdafx.h"


using namespace m::alloc;
using namespace m::error;

#ifdef _DEBUG
//#define DBG_LOCK
#endif

namespace m
{
#ifdef DBG_LOCK
	ATOMIC_ALIGN static volatile LONG ExclusiveLocks = 0;
	ATOMIC_ALIGN static volatile LONG SharedLocks = 0;

	static inline void DbgLock(LPCWSTR LockName, LPCWSTR File, DWORD Line)
	{
		WCHAR tid[4096] = {};
		StringCchPrintf(tid, ARRAYSIZE(tid), L"%s:%d %s - tid %d sh %d ex %d\n", File, Line, LockName, GetCurrentThreadId(), SharedLocks, ExclusiveLocks);
		OutputDebugString(tid);
	}
#else
	static inline void DbgLock(LPCWSTR, LPCWSTR File, DWORD Line)
	{

	}
#endif

	static inline void AcquireSRWLockShared(PSRWLOCK lock, LPCWSTR File, DWORD Line)
	{
		DbgLock(L"AcquireSRWLockShared", File, Line);
		::AcquireSRWLockShared(lock);
#ifdef DBG_LOCK
		InterlockedIncrement(&SharedLocks);
#endif // DBG_LOCK
	}

	static inline void AcquireSRWLockExclusive(PSRWLOCK lock, LPCWSTR File, DWORD Line)
	{
		DbgLock(L"AcquireSRWLockExclusive", File, Line);
		::AcquireSRWLockExclusive(lock);
#ifdef DBG_LOCK
		InterlockedIncrement(&ExclusiveLocks);
#endif // DBG_LOCK
	}

	static inline void ReleaseSRWLockShared(PSRWLOCK lock, LPCWSTR File, DWORD Line)
	{
		DbgLock(L"ReleaseSRWLockShared", File, Line);
		::ReleaseSRWLockShared(lock);
#ifdef DBG_LOCK
		InterlockedDecrement(&SharedLocks);
#endif // DBG_LOCK
	}

	static inline void ReleaseSRWLockExclusive(PSRWLOCK lock, LPCWSTR File, DWORD Line)
	{
		DbgLock(L"ReleaseSRWLockExclusive", File, Line);
		::ReleaseSRWLockExclusive(lock);
#ifdef DBG_LOCK
		InterlockedDecrement(&ExclusiveLocks);
#endif // DBG_LOCK
	}

	static inline void WakeAllConditionVariable(PCONDITION_VARIABLE ConditionVariable, LPCWSTR File, DWORD Line)
	{
		DbgLock(L"WakeAllConditionVariable", File, Line);
		::WakeAllConditionVariable(ConditionVariable);
	}

	static inline BOOL SleepConditionVariableSRW(
		PCONDITION_VARIABLE ConditionVariable,
		PSRWLOCK SRWLock,
		DWORD dwMilliseconds,
		ULONG Flags, 
		LPCWSTR File, DWORD Line)
	{
#ifdef DBG_LOCK
		InterlockedDecrement(&SharedLocks);
#endif // DBG_LOCK
		DbgLock(L"SleepConditionVariableSRW", File, Line);
		BOOL ret = ::SleepConditionVariableSRW(ConditionVariable, SRWLock, dwMilliseconds, Flags);
#ifdef DBG_LOCK
		InterlockedIncrement(&SharedLocks);
#endif // DBG_LOCK
		return ret;
	}

	static inline void EnterCriticalSection(LPCRITICAL_SECTION lpCriticalSection, LPCWSTR File, DWORD Line)
	{
		DbgLock(L"EnterCriticalSection", File, Line);
		::EnterCriticalSection(lpCriticalSection);
	}

	static inline void LeaveCriticalSection(LPCRITICAL_SECTION lpCriticalSection, LPCWSTR File, DWORD Line)
	{
		DbgLock(L"LeaveCriticalSection", File, Line);
		::LeaveCriticalSection(lpCriticalSection);
	}

	process_manager::process_manager()
		: _FirstProcess(nullptr),
		_LastMessage(nullptr),
		_NumProcesses(0),
		_Started(false),
		_Running(false),
		_Sent(0),
		_TimerQ(CreateTimerQueue()),
		_Stopped(0)
	{
		InitializeSRWLock(&_MessageLock);
		InitializeConditionVariable(&_MessageEvent);
		if (!InitializeCriticalSectionAndSpinCount(&_ProcessLock, CRIT_WAIT))
		{
			throw bad_result_exception(HRESULT_FROM_WIN32(GetLastError()), "Failed to initialize process lock.");
		}
		if (!_TimerQ)
		{
			throw bad_result_exception(HRESULT_FROM_WIN32(GetLastError()), "Process manager couldn't be started.");
		}
	}

	process_manager::~process_manager()
	{
		DeleteCriticalSection(&_ProcessLock);

		while (_FirstProcess)
		{
			assign(_FirstProcess, _FirstProcess->_NextProcess);
		}
		
		clear(_LastMessage);

		if (_TimerQ)
		{
			if (!DeleteTimerQueueEx(_TimerQ, INVALID_HANDLE_VALUE))
			{
				DBGW(GetLastError(), L"Failed to delete timer queue.");
			}
		}
	}

	void process_manager::send(process* sender, const wchar_t* message)
	{
		ref_storage track1;
		m::message* m = create(track1, new m::message(sender, message));
		Send(m, false);
	}

	void process_manager::send(process* sender, const wchar_t* message, object* data)
	{
		ref_storage track1;
		m::message* m = create(track1, new m::message(sender, message, data));
		Send(m, false);
	}

	void process_manager::send(process* sender, const wchar_t* message, object* data, bool cancelLast)
	{
		ref_storage track1;
		m::message* m = create(track1, new m::message(sender, message, data));
		Send(m, cancelLast);
	}

	void process_manager::Send(message* msg, bool cancelLast)
	{
		// To queue a message for sending, we add it to the last message on the list.
		// Individual processes then keep track of the last message they recieve. 
		// As each process advances to the last message, it decrements the reference count we set here.
		// Finally, we decrement our reference count on the last message when we advance to the nextone.


#ifndef _DISABLE_MONO_DEBUGGING
		dbg.message(msg->type(), msg->sender(), msg->data());
#endif // _DISABLE_MONO_DEBUGG

		bool wakeAll = false;
		m::AcquireSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		if (_NumProcesses == 0)
		{
			// We could leak memory if no process is available to read the given message.
			m::ReleaseSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
			return;
		}

		if (cancelLast)
		{
			// Cancel the last sent message.
			// We have to find a link to the earliest message. This actually involves going through each process
			// to find the message with the lowest sequence number, then starting with that, searching through
			// the list to find the first message matching our id. 
			// This operation is linear in time.

			message* FirstMessage = FindFirstMessage();

			if (FirstMessage)
			{
				message* cur = FirstMessage->_NextMessage;

				// Now we found the first message, we need to go through the message list and remove it.

				while (cur)
				{
					if (cur->type_equals(msg->type()))
					{
						if (!cur->_Canceled)
						{
							cur->_Canceled = true;
							break;
						}
					}
					cur = cur->_NextMessage;
				}
			}
		}

		// Add extra references equal to the number of subscribed processes. When each process reads the object, it will release it held reference.
		msg->reference(_NumProcesses);
		++_Sent;

		if (_LastMessage)
		{
			assign(_LastMessage->_NextMessage, msg);
		}
		assign(_LastMessage, msg);
		if (_Running)
		{
			wakeAll = true;
		}

		m::ReleaseSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		if (wakeAll)
		{
			m::WakeAllConditionVariable(&_MessageEvent, __FILEW__, __LINE__);
		}
	}

	message* process_manager::FindFirstMessage()
	{
		int maxDiff = INT32_MIN;
		process* curProc = _FirstProcess;
		message* FirstMessage = nullptr;

		while (curProc)
		{
			// There's a few cases involved with determining if this message is a predecessor.
			// a) The previous value is null.
			// b) The previous value is the maximum value of Message::IdType and this value isn't.
			// c) The previous value is less than this value.

			if (curProc->_CurMsg)
			{
				if (FirstMessage)
				{
					int diff = _Sent - curProc->_Read;
					if (diff < 0)
					{
						diff = -diff;
					}
					if (diff > maxDiff)
					{
						FirstMessage = curProc->_CurMsg;
						maxDiff = diff;
					}
				}
				else
				{
					FirstMessage = curProc->_CurMsg;
				}
			}
			curProc = curProc->_NextProcess;
		}

		return FirstMessage;
	}

	void process_manager::start()
	{
		m::EnterCriticalSection(&_ProcessLock, __FILEW__, __LINE__);

		m::AcquireSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		if (!_Started)
		{
			_Started = _Running = true;
		}
		m::ReleaseSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
	}

	void process_manager::stop()
	{
		m::EnterCriticalSection(&_ProcessLock, __FILEW__, __LINE__);

		if (!_Started)
		{
			m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
			throw bad_result_exception(MONORET(FACILITY_PROCESS, ERROR_INVALID_OPERATION), "The process manager wasn't running.");
		}

		if (!_Running)
		{
			m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
			return;
		}
		
		_Stopped = 0;
		size_t countProcesses = 0;
		process* cur = _FirstProcess;
		while (cur)
		{
			++countProcesses;
			cur = cur->_NextProcess;
		}

		send(nullptr, L"us.sl.m.stop", nullptr);

		while (_Stopped < countProcesses)
		{
			m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
			SleepEx(0, FALSE);
			m::EnterCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
		}

		m::AcquireSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		_Started = _Running = false;
		m::ReleaseSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);

		m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
	}

	void process_manager::pause()
	{
		m::EnterCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
		m::AcquireSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		_Running = false;
		m::ReleaseSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
	}

	void process_manager::resume()
	{
		m::EnterCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
		bool shouldWake = false;
		m::AcquireSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		if (!_Running)
		{
			_Running = true;
			shouldWake = true;
		}
		m::ReleaseSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		if (shouldWake)
		{
			m::WakeAllConditionVariable(&_MessageEvent, __FILEW__, __LINE__);
		}
		m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
	}

	const bool& process_manager::started() const
	{
		return _Started;
	}

	const bool& process_manager::resumed() const
	{
		return _Running;
	}

	void process_manager::add_process(process * process)
	{
		if (!_Started)
		{
			throw bad_result_exception(MONORET(FACILITY_PROCESS, ERROR_INVALID_OPERATION), "The process manager must be started before it can accept processes.");
		}

		assert(process);
		if (process)
		{
			bool wakeAll = false;

			m::EnterCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
			m::AcquireSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
			assign(process->_NextProcess, _FirstProcess);
			if (_FirstProcess)
			{
				assign(_FirstProcess->_LastProcess, process);
			}
			assign(_FirstProcess, process);
			++_NumProcesses;
			ref_storage track1;
			m::message* msg = create(track1, new m::message(nullptr, L"us.sl.m.start", process));
			// Add extra references equal to the number of subscribed processes. When each process reads the object, it will release it held reference.
			msg->reference(_NumProcesses);
			++_Sent;

			if (_LastMessage)
			{
				assign(_LastMessage->_NextMessage, msg);
			}
			assign(_LastMessage, msg);
			if (_Running)
			{
				wakeAll = true;
			}
			m::ReleaseSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);

			ref_storage track2;
			weak_ref* wr = create_weak(track2, this);
			process->Start(wr, msg);
			m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);

			if (wakeAll)
			{
				m::WakeAllConditionVariable(&_MessageEvent, __FILEW__, __LINE__);
			}
		}
	}

	void process_manager::remove_process(process * process)
	{
		assert(process);
		if (process)
		{
			ref_storage track1;
			weak_ref* weak = create_weak(track1, this);

			if (process->_PmWeak != weak)
			{
				// We were asked to remove a process we don't explicitly own.
				throw bad_result_exception(MONORET(FACILITY_PROCESS, ERROR_NOT_FOUND), "Process not found.");
			}

			send(nullptr, L"us.sl.m.stop", process);
			while (process->has_message())
			{
				SleepEx(0, TRUE);
			}
			process->Stop();
		}
	}

	void process_manager::flush()
	{
		m::EnterCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
		process* cur = _FirstProcess;

		while (cur)
		{
			while (cur->has_message())
			{
				SleepEx(0, TRUE);
			}

			cur = cur->_NextProcess;
		}

		m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
	}

	void process_manager::Remove(process* process)
	{
		m::EnterCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
		m::AcquireSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);

		if (process->_NextProcess)
		{
			assign(process->_NextProcess->_LastProcess, process->_LastProcess);
		}
		if (process->_LastProcess)
		{
			assign(process->_LastProcess->_NextProcess, process->_NextProcess);
		}
		if (process == _FirstProcess)
		{
			assign(_FirstProcess, process->_NextProcess);
		}

		assert(_NumProcesses > 0);
		if (_NumProcesses > 0)
		{
			--_NumProcesses;
		}
		
		++_Stopped;

		m::ReleaseSRWLockExclusive(&_MessageLock, __FILEW__, __LINE__);
		m::LeaveCriticalSection(&_ProcessLock, __FILEW__, __LINE__);
	}

	void process_manager::ChangeTimer(timer_state** Ts, const scalar& delay, const scalar& interval, const bool& oneShot, process* Sender, const wchar_t* evt, object* data)
	{
		if (!(*Ts))
		{
			assign_new(*Ts, new timer_state());
		}

		(*Ts)->CreateOrSet(
			this,
			abs(delay),
			abs(interval),
			oneShot,
			Sender,
			evt,
			data
		);
	}

	void process_manager::ClearTimer(timer_state** Ts)
	{
		if (*Ts)
		{
			ref_storage track1;
			weak_ref* weak = create_weak(track1, this);
			if ((*Ts)->_PmWeak != weak)
			{
				// We were asked to remove a process we don't explicitly own.
				throw bad_result_exception(MONORET(FACILITY_PROCESS, ERROR_NOT_FOUND), "Process not found.");
			}

			(*Ts)->clear();
			clear(*Ts);
		}
	}

	bool process_manager::GetNext(message** msg)
	{
		assert(msg);
		if (msg)
		{
			m::AcquireSRWLockShared(&_MessageLock, __FILEW__, __LINE__);

			if (_Running)
			{
				// We use the following logic to atomically get the next message.
				// a) if the message target is null, by definition, the message target is the last message to be sent.
				// b) if the message target is not null, update the message target to the next one.

				assert(*msg);
				if (*msg)
				{
					message* last = *msg;
					message* next = (*msg)->_NextMessage;
					if (next)
					{
						(*msg)->release();
						*msg = next;
					}

					if (last != *msg)
					{
						if (!(*msg)->_Canceled)
						{
							m::ReleaseSRWLockShared(&_MessageLock, __FILEW__, __LINE__);
							return true;
						}
						else
						{
							m::ReleaseSRWLockShared(&_MessageLock, __FILEW__, __LINE__);
							return false;
						}
					}
					else
					{
						m::SleepConditionVariableSRW(&_MessageEvent, &_MessageLock, INFINITE, CONDITION_VARIABLE_LOCKMODE_SHARED, __FILEW__, __LINE__);
						m::ReleaseSRWLockShared(&_MessageLock, __FILEW__, __LINE__);
						return false;
					}
				}
			}
		}
		return false;
	}

	bool process_manager::HasMessageAbove(const message* msg) const
	{
		assert(msg);
		if (msg)
		{
			m::AcquireSRWLockShared(&_MessageLock, __FILEW__, __LINE__);

			// We use the following logic to atomically get the next message.
			// a) if the message target is null, by definition, the message target is the last message to be sent.
			// b) if the message target is not null, update the message target to the next one.

			if (msg)
			{
				do
				{
					if (!msg->_NextMessage)
					{
						m::ReleaseSRWLockShared(&_MessageLock, __FILEW__, __LINE__);
						return false;
					}
					msg = msg->_NextMessage;
				} while (msg->_Canceled);

				m::ReleaseSRWLockShared(&_MessageLock, __FILEW__, __LINE__);
				return true;
			}
			else if (_LastMessage)
			{
				m::ReleaseSRWLockShared(&_MessageLock, __FILEW__, __LINE__);
				// msg is null. This process just started.
				return true;
			}

			m::ReleaseSRWLockShared(&_MessageLock, __FILEW__, __LINE__);
		}

		return false;
	}
}